package com.shujia.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

public class WordCount {


    /**
     * map端
     * 执行输入输出key value的类型
     * <p>
     * LongWritable  相当于Long
     * Text 相当于String
     */
    static class WordCountMao extends Mapper<LongWritable, Text, Text, LongWritable> {

        /**
         * map  方法 一行数据都会被执行一次
         *
         * @param key     数据行号   偏移量
         * @param value   一行数据
         * @param context 上下文对象    将map的数据写入到reduce端
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            //获取一行数据 返回一个字符串
            String line = value.toString();

            String[] split = line.split(",");

            for (String word : split) {

                //构建输出的key
                Text wordKey = new Text(word);

                //value
                LongWritable intValue = new LongWritable(1);

                //将数据写入reduce    每一次输出，相当于输出一行数据
                context.write(wordKey, intValue);
            }
        }
    }

    /**
     * reduce端 对相同的key的value进行聚合操作
     * <p>
     * 执行  输入key  value的类型  类型必须和map端输出类型相同
     */

    static class WordCountReduce extends Reducer<Text, LongWritable, Text, LongWritable> {

        /**
         * reduce  每一个key调用一次
         *
         * @param key     单词
         * @param values  同一个单词的多个value   一堆1
         * @param context 上下文对象
         */

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

            /**
             * 统计单词数量
             *
             */

            Long sum = 0L;

            for (LongWritable value : values) {
                sum = sum + value.get();
            }


            /**
             * 将数据写入hdfs
             */
            LongWritable value = new LongWritable(sum);
            context.write(key, value);

        }
    }


    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        /**
         * 构建mapreduce任务
         * 1、指定map端  执行reduce
         * 2、指定输入输出路径
         *
         */

        /**
         * 创建hadoop配置文件对象
         * 如果成行在集群里面通过hadoop jar的方式提交   会自动加载集群里面hadoop配置文件
         *
         */
        Configuration conf = new Configuration();

        /**
         * 设置 Splitsize大小
         * Math.max(minSize, Math.min(maxSize, blockSize));
         */
        conf.set("mapreduce.input.fileinputformat.split.maxsize", "50");

        // 构建mapreduce任务对象
        Job job = Job.getInstance(conf);


        //指定reducetask 数量  默认值是1
        /**
         * reducetask数量设置
         * 1、不能太多，太多了会产生很多小文件，hdfs压力会很大
         * 2、不能太少，太少，任务执行很慢
         *
         * 保证每个reduce处理的数据在100m-1G  比较合适
         *
         */
        job.setNumReduceTasks(2);

        //指定任务名   可以在yarn里面查看
        job.setJobName("wordcount");

        //指定主类   main函数所在的类
        job.setJarByClass(WordCount.class);

        //指定map端
        job.setMapperClass(WordCountMao.class);


        /**
         * 为什么要指定输入输出类型
         * 泛型只在编译时起作用
         * 反射在运行时起作用
         *
         */

        //指定map输出keyvalue类型 和上面map类里面指定的一样
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);


        //指定reduce端
        job.setReducerClass(WordCountReduce.class);

        //执行输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //指定输入输出路径
        TextInputFormat.addInputPath(job, new Path("/words.txt"));


        /**
         * 删除输出目录
         */
        Path outPath = new Path("/data/out");
        FileSystem fileSystem = FileSystem.get(conf);
        if (fileSystem.exists(outPath)) {
            fileSystem.delete(outPath, true);
        }


        //输出路径是一个目录
        TextOutputFormat.setOutputPath(job, outPath);

        //等待任务执行完成
        job.waitForCompletion(true);

        /**
         *
         * 提交任务
         *
         * 启动hdfs和yarn
         *
         * 1、打包上传
         * 2、提交执行
         * hadoop jar hadoop-1.0-SNAPSHOT.jar com.shujia.mapreduce.WordCount
         */


    }
}
